package com.yanggu.flink.datastream_api.multi_stream_transform.combine_stream

import com.yanggu.flink.datastream_api.pojo.Event
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * 基于间隔的join IntervalJoin
 * IntervalJoin>>>>> (Alice,order-2,5000)=>Event(Alice,./prod?id=100,3000)
 * IntervalJoin>>>>> (Alice,order-2,5000)=>Event(Alice,./prod?id=200,3500)
 * IntervalJoin>>>>> (Alice,order-2,5000)=>Event(Alice,./prod?id=300,3600)
 */
object IntervalJoinDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //订单流数据
    val orderDataStream = env
      .fromElements(
        ("Mary", "order-1", 5000L),
        ("Alice", "order-2", 5000L),
        ("Bob", "order-3", 20000L),
        ("Alice", "order-4", 20000L),
        ("Cary", "order-5", 51000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, String, Long)] {
          override def extractTimestamp(element: (String, String, Long), recordTimestamp: Long) = element._3
        })
      )

    //点击流数据
    val clickDataStream = env
      .fromElements(
        Event("Bob", "./cart", 2000L),
        Event("Alice", "./prod?id=100", 3000L),
        Event("Alice", "./prod?id=200", 3500L),
        Event("Bob", "./prod?id=2", 2500L),
        Event("Alice", "./prod?id=300", 3600L),
        Event("Bob", "./home", 3000L),
        Event("Bob", "./prod?id=1", 2300L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[Event] {
          override def extractTimestamp(element: Event, recordTimestamp: Long) = element.timestamp
        })
      )

    orderDataStream
      .keyBy(data => data._1)
      .intervalJoin(clickDataStream.keyBy(data => data.name))
      //between是包含上边界和下边界的
      .between(Time.seconds(-5L), Time.seconds(10L))
      //去除上边界
      //.lowerBoundExclusive()
      //去除下边界
      .upperBoundExclusive()
      .process((left: (String, String, Long),
                right: Event,
                _: ProcessJoinFunction[(String, String, Long), Event, String]#Context,
                out: Collector[String]) => {
        out.collect(left + "=>" + right)
      })
      .print("IntervalJoin>>>>")

    env.execute()

  }

}
